package com.mgm.kafka.service.impl;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;

import com.mgm.kafka.service.KafkaProducerService;

@Service
public class KafkaProducerServiceImpl implements KafkaProducerService {
	
	private static Logger logger = LoggerFactory.getLogger(KafkaProducerServiceImpl.class);
	
	private static final int PARTITION_NUM = 0;
	
	@Autowired
    private KafkaTemplate<String, String> kafkaTemplate; 
    
	@Override
	public void sendMessageDefault(String key, String data) {
		sendMessage("topic07101001", 0, null, key, data);
	}
	
	public void sendMessage(String topic, Integer partition, Long timestamp,  String key, String data) {
        ListenableFuture<SendResult<String, String>> listenableFuture = null;
        if (kafkaTemplate.getDefaultTopic().equals(topic)) {
            listenableFuture =  kafkaTemplate.sendDefault(partition,timestamp,key,data);
        }else {
            listenableFuture = kafkaTemplate.send(topic, partition, timestamp, key, data);
        }
        //发送成功回调
        SuccessCallback<SendResult<String, String>> successCallback = new SuccessCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> result) {
                //成功业务逻辑
            	logger.info("[KafkaProducer callback] success...");
                System.out.println("成功");
            }
        };
        //发送失败回调
        FailureCallback failureCallback = new FailureCallback() {
            @Override
            public void onFailure(Throwable ex) {
                //失败业务逻辑
            	logger.info("[KafkaProducer callback] fail...",ex.getMessage());
                throw new RuntimeException(ex);
            }
        };
        listenableFuture.addCallback(successCallback, failureCallback);
    }
	
	public void sendMessage(String topic, String data) {
		logger.debug("The message is to be send by kafka is : topic = {}, data = {}", topic, data);
		if(!StringUtils.isEmpty(data)) {
			sendMessage(topic, PARTITION_NUM, null, "key", data);
		}
	}

/*	public KafkaTemplate<String, String> getKafkaTemplate() {
		return kafkaTemplate;
	}

	public void setKafkaTemplate(KafkaTemplate<String, String> kafkaTemplate) {
		this.kafkaTemplate = kafkaTemplate;
	}*/

}